/*
 *
 *  * | Licensed 未经许可不能去掉「OPENIITA」相关版权
 *  * +----------------------------------------------------------------------
 *  * | Author: xw2sy@163.com
 *  * +----------------------------------------------------------------------
 *
 *  Copyright [2024] [OPENIITA]
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 * /
 */

package cc.iotkit.plugins.tcp.server;


import cc.iotkit.common.exception.BizException;
import cc.iotkit.common.utils.HexUtil;
import cc.iotkit.plugin.core.IPluginScript;
import cc.iotkit.plugin.core.thing.IThingService;
import cc.iotkit.plugin.core.thing.actions.ActionResult;
import cc.iotkit.plugin.core.thing.actions.DeviceState;
import cc.iotkit.plugin.core.thing.actions.up.*;
import cc.iotkit.plugins.tcp.cilent.VertxTcpClient;
import cc.iotkit.plugins.tcp.conf.DeviceInfoConfig;
import cc.iotkit.plugins.tcp.conf.TcpServerConfig;
import cc.iotkit.plugins.tcp.parser.*;
import cc.iotkit.script.IScriptEngine;
import cn.hutool.core.util.IdUtil;
import com.gitee.starblues.bootstrap.annotation.AutowiredType;
import com.gitee.starblues.core.PluginInfo;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetServerOptions;
import io.vertx.core.net.NetSocket;
import io.vertx.core.parsetools.RecordParser;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 *
 *
 */
@Slf4j
@Service
public class TcpServerVerticle extends AbstractVerticle {

    @Getter
    @Setter
    private TcpServerConfig tcpServerConfig;

    @Getter
    @Setter
    private DeviceInfoConfig deviceInfoConfig;

    private final Map<String, VertxTcpClient> clientMap = new ConcurrentHashMap<>();

    private final Map<String, String> dnToPk = new HashMap<>();

    private final Map<String, Long> heartbeatDevice = new HashMap<>();

    @Setter
    private long keepAliveTimeout = Duration.ofSeconds(30).toMillis();

    private NetServer netServer;

    @Getter
    private IScriptEngine scriptEngine;

    @Autowired
    private PluginInfo pluginInfo;

    @Autowired
    @AutowiredType(AutowiredType.Type.MAIN_PLUGIN)
    private IPluginScript pluginScript;

    @Autowired
    @AutowiredType(AutowiredType.Type.MAIN_PLUGIN)
    private IThingService thingService;

    @Override
    public void stop() {
        if (netServer != null) {
            netServer.close(rst -> {
                log.info("TcpServerVerticle tcp server close: {}", rst.succeeded());
            });
        }

        log.info("TcpServerVerticle tcp server stopped");
    }

    /**
     * 创建配置文件
     */
    @PostConstruct
    public void initConfig() {
        log.info("TcpServerVerticle initConfig: {}", pluginScript.getClass().getName());
        //获取脚本引擎
        scriptEngine = pluginScript.getScriptEngine(pluginInfo.getPluginId());
        if (scriptEngine == null) {
            throw new BizException("TcpServerVerticle script engine is null");
        }
        Executors.newSingleThreadScheduledExecutor().schedule(this::initTcpServer, 3, TimeUnit.SECONDS);
    }


    /**
     * 初始TCP服务
     */
    public void initTcpServer() {
        log.info("TcpServerVerticle initTcpServer start");
        netServer = vertx.createNetServer(new NetServerOptions().setHost(tcpServerConfig.getHost()).setPort(tcpServerConfig.getPort()));
        netServer.connectHandler(this::acceptTcpConnection);
        netServer.listen(tcpServerConfig.createSocketAddress(), result -> {
            if (result.succeeded()) {
                log.info("TcpServerVerticle initTcpServer tcp server startup on {}", result.result().actualPort());
            } else {
                result.cause().printStackTrace();
            }
        });
    }

    public void sendMsg(String clientId, Buffer msg) {
        log.info("TcpServerVerticle sendMsg start clientId = {}  msg = {}", clientId , msg);
        VertxTcpClient tcpClient = clientMap.get(clientId);
        if (tcpClient != null) {
            tcpClient.sendMessage(msg);
        }
        log.info("TcpServerVerticle sendMsg end clientId = {}, msg = {}", clientId, msg);
    }

    @Scheduled(fixedRate = 40, timeUnit = TimeUnit.SECONDS)
    public void offlineCheckTask() {
        //log.info("TcpServerVerticle offlineCheckTask start");
        Set<String> clients = new HashSet<>(clientMap.keySet());
        for (String clientId : clients) {
            VertxTcpClient client = clientMap.get(clientId);
            if (!client.isOnline()) {
                log.info("TcpServerVerticle offlineCheckTask clientId = {}", client.getId());
                client.shutdown();
            }
        }
        //log.info("TcpServerVerticle offlineCheckTask start1");
        heartbeatDevice.keySet().iterator().forEachRemaining(clientId -> {
            //log.info("TcpServerVerticle offlineCheckTask clientId = {}", Arrays.toString(clientId.getBytes()));
            Long time = heartbeatDevice.get(clientId);
            //心跳超时，判定为离线
            if (System.currentTimeMillis() - time > keepAliveTimeout * 2) {
                heartbeatDevice.remove(clientId);

                DeviceStateChange deviceStateChange = DeviceStateChange.builder()
                        .id(IdUtil.simpleUUID())
                        .productKey(dnToPk.get(clientId))
                        .deviceName(clientId)
                        .state(DeviceState.OFFLINE)
                        .time(System.currentTimeMillis())
                        .build();

                log.info("TcpServerVerticle offlineCheckTask pluginInfo.getPluginId() = {}", pluginInfo.getPluginId());
                log.info("TcpServerVerticle offlineCheckTask productKey = {}", dnToPk.get(clientId));
                log.info("TcpServerVerticle offlineCheckTask deviceStateChange = {}", deviceStateChange.toString());

                //离线上报
                thingService.post(pluginInfo.getPluginId(), deviceStateChange);
            }
        });
    }

    /**
     * TCP连接处理逻辑
     *
     * @param socket socket
     */
    public void acceptTcpConnection(NetSocket socket) {
        // 客户端连接处理
        //String clientId = socket.remoteAddress().toString();
        //String productKey = IdUtil.simpleUUID();
        String deviceId = deviceInfoConfig.getDeviceId();
        String productKey = deviceInfoConfig.getProductKey();
        String deviceName = deviceInfoConfig.getDeviceName();
        dnToPk.put(deviceId, productKey);

        log.info("TcpServerVerticle acceptTcpConnection NetSocket deviceId = {}", deviceId);
        VertxTcpClient client = new VertxTcpClient(deviceId);
        try {
            // TCP异常和关闭处理
            socket.exceptionHandler(Throwable::printStackTrace).closeHandler(nil -> {
                log.info("TcpServerVerticle acceptTcpConnection tcp server client [{}] closed", socket.remoteAddress());
                client.shutdown();
            });
            // 这个地方是在TCP服务初始化的时候设置的 parserSupplier
            client.setKeepAliveTimeoutMs(keepAliveTimeout);
            client.setSocket(socket);
            if(!clientMap.containsKey(deviceId)) {
                clientMap.put(deviceId, client);
            }
            log.info("TcpServerVerticle acceptTcpConnection tcp server client [{}] accept", socket.remoteAddress());

            RecordParser parser = DataReader.getParser(buffer -> {
                try {
//                    if (log.isDebugEnabled()) {
//                        log.info("TcpServerVerticle acceptTcpConnection buffer.length:[{}}] buffer:[{}}]",
//                                buffer.length(), HexUtil.toHexString(buffer.getBytes()));
//                    }else {
//                        System.out.printf("TcpServerVerticle acceptTcpConnection buffer.length:[%d] buffer:[%s]",
//                                buffer.length(), HexUtil.toHexString(buffer.getBytes()));
//                    }
                    DataRevPackage data = DataDecoder.decode(buffer);
                    byte cmd = data.getCmd();
                    byte code = data.getCode();
//                    if (log.isDebugEnabled()) {
//                        log.info("TcpServerVerticle acceptTcpConnection data.cmd:[{}]  data.code:[{}] data.payload:[{}]",
//                                cmd, code, HexUtil.toHexString(data.getPayload()));
//                    }else {
//                        System.out.printf("TcpServerVerticle acceptTcpConnection data.cmd:[%d]  data.code:[%d] data.payload:[%s]",
//                                cmd, code, HexUtil.toHexString(data.getPayload()));
//                    }
                    switch (cmd){
                        case DataRevPackage.CMD_QUERY_SYSTEM:
                            //查询设备参数
                            Map<String, Object> paramsMap = new HashMap<>();
                            paramsMap.put("cmd",data.getCmd());
                            paramsMap.put("code",data.getCode());
                            paramsMap.put("length",data.getLength());
                            paramsMap.put("payload",data.getPayload());

                            DeviceQuery deviceQuery = DeviceQuery.builder()
                                    .id(IdUtil.simpleUUID())
                                    .deviceId(deviceId)
                                    .deviceName(deviceName)
                                    .productKey(productKey)
                                    .params(paramsMap)
                                    .time(System.currentTimeMillis())
                                    .build();

                            if (log.isDebugEnabled()) {
                                log.info("TcpServerVerticle acceptTcpConnection deviceQuery = [{}]", deviceQuery.toString());
                            }else {
                                System.out.printf("TcpServerVerticle acceptTcpConnection deviceQuery = [%s]", deviceQuery.toString());
                            }

                            ActionResult actionResult = thingService.post(pluginInfo.getPluginId(), deviceQuery);
                            if (actionResult.getCode() != 0) {
                                log.error("TcpServerVerticle acceptTcpConnection post error! actionResult = {}", actionResult);
                            }
                            break;
                        case DataRevPackage.CMD_SET_PARAM:
                            //查询设备参数
                            Map<String, Object> paramsMap1 = new HashMap<>();
                            paramsMap1.put("cmd",data.getCmd());
                            paramsMap1.put("code",data.getCode());
                            paramsMap1.put("length",data.getLength());
                            paramsMap1.put("payload",data.getPayload());

                            DeviceParamChange deviceParamChange = DeviceParamChange.builder()
                                    .id(IdUtil.simpleUUID())
                                    .deviceId(deviceId)
                                    .deviceName(deviceName)
                                    .productKey(productKey)
                                    .params(paramsMap1)
                                    .time(System.currentTimeMillis())
                                    .build();

                            if (log.isDebugEnabled()) {
                                log.info("TcpServerVerticle acceptTcpConnection deviceParamChange = [{}]", deviceParamChange.toString());
                            }else {
                                System.out.printf("TcpServerVerticle acceptTcpConnection deviceParamChange = [%s]", deviceParamChange.toString());
                            }

                            ActionResult actionResult1 = thingService.post(pluginInfo.getPluginId(), deviceParamChange);
                            if (actionResult1.getCode() != 0) {
                                log.error("TcpServerVerticle acceptTcpConnection post error! actionResult1 = {}", actionResult1);
                            }
                            break;
                        case DataRevPackage.CMD_DATA_COLLECT:
                            //Map<String, Object> paramsMap2 = scriptEngine.invokeMethod(new TypeReference<>() {}, "decode", data);
                            Map<String, Object> paramsMap2 = new HashMap<>();
                            paramsMap2.put("cmd",data.getCmd());
                            paramsMap2.put("code",data.getCode());
                            paramsMap2.put("length",data.getLength());
                            paramsMap2.put("payload",data.getPayload());

                            //采集数据上报
                            DeviceCollectReport collectReport = DeviceCollectReport.builder()
                                    .id(IdUtil.simpleUUID())
                                    .deviceId(deviceId)
                                    .deviceName(deviceName)
                                    .productKey(productKey)
                                    .params(paramsMap2)
                                    .time(System.currentTimeMillis())
                                    .build();

//                            if (log.isDebugEnabled()) {
//                                log.info("TcpServerVerticle acceptTcpConnection collectReport = [{}]", collectReport.toString());
//                            }else {
//                                System.out.printf("TcpServerVerticle acceptTcpConnection collectReport = [%s]", collectReport.toString());
//                            }

                            ActionResult actionResult2 = thingService.post(pluginInfo.getPluginId(), collectReport);
                            if (actionResult2.getCode() != 0) {
                                log.error("TcpServerVerticle acceptTcpConnection post error! actionResult2 = {}", actionResult2);
                            }
                            break;
                        default:
                            //心跳
                            online(deviceId, deviceName);
                            heartbeatDevice.put(deviceId, System.currentTimeMillis());
                            log.error("TcpServerVerticle acceptTcpConnection DataPackage error cmd = {}", cmd);
                            break;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    //非法数据断开连接
                    log.error("TcpServerVerticle acceptTcpConnection DataPackage error e = {}", e.getMessage());
                    socket.close();
                }
            });

            client.setParser(parser);
            if (log.isDebugEnabled()) {
                log.info("TcpServerVerticle acceptTcpConnection client [{}] connection", socket.remoteAddress());
            }else {
                System.out.printf("TcpServerVerticle acceptTcpConnection client [%s] connection", socket.remoteAddress());
            }
        } catch (Exception e) {
            e.printStackTrace();
            client.shutdown();
        }
    }

    public void online(String deviceId, String deviceName) {
        if (!heartbeatDevice.containsKey(deviceId)) {
            //第一次心跳，上线
            DeviceStateChange deviceStateChange = DeviceStateChange.builder()
                    .id(IdUtil.simpleUUID())
                    .deviceId(deviceId)
                    .deviceName(deviceName)
                    .productKey(dnToPk.get(deviceId))
                    .state(DeviceState.ONLINE)
                    .time(System.currentTimeMillis())
                    .build();
            if (log.isDebugEnabled()) {
                log.info("TcpServerVerticle online deviceId = [{}] deviceStateChange = [{}]", deviceId, deviceStateChange.toString());
            }else {
                System.out.printf("TcpServerVerticle online deviceId = [%s] deviceStateChange = [%s]", deviceId, deviceStateChange.toString());
            }
            thingService.post(pluginInfo.getPluginId(), deviceStateChange);
        }
    }
}
